package com.learn.window;

import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.List;

/**
 * @create: 2023-04-20 09:42
 * @author: Mr.Du
 * --------------
 * @notes: process方法做全量+增量聚合
 * 全量聚合：窗口需要维护全部原始数据，窗口触发进行全量聚合
 **/
public class ProcessWindowFunctionDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        List<Tuple3<String,String,Long>> ENGLISH = new ArrayList<>();
        ENGLISH.add(Tuple3.of("class1", "张三", 100L));
        ENGLISH.add(Tuple3.of("class1", "李四", 40L));
        ENGLISH.add(Tuple3.of("class1", "王五", 60L));
        ENGLISH.add(Tuple3.of("class2", "赵六", 20L));
        ENGLISH.add(Tuple3.of("class2", "小七", 30L));
        ENGLISH.add(Tuple3.of("class2", "小八", 50L));


        DataStreamSource<Tuple3<String, String, Long>> source = env.fromCollection(ENGLISH);

        source
                .keyBy(0)
                .countWindow(2).process(new ProcessWindowFunction<Tuple3<String, String, Long>, Double, Tuple, GlobalWindow>() {
                    @Override
                    public void process(Tuple tuple, ProcessWindowFunction<Tuple3<String, String, Long>, Double, Tuple, GlobalWindow>.Context context, Iterable<Tuple3<String, String, Long>> iterable, Collector<Double> out) throws Exception {
                        long sum = 0;
                        long count = 0;
                        for (Tuple3<String,String,Long> in:iterable
                             ) {
                            sum+=in.f2;
                            count++;
                        }
                        out.collect((double)(sum/count));
                    }
                }).print("processFunction>>>");
        
        env.execute();
    }
}
